package qa.d20241106;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * version: v0.4
 *  窗口join
 */
public class JoinTest4 {


  public static void main(String[] args) throws InterruptedException {


    Configuration flinkConf = new Configuration();
    flinkConf.setString("rest.port","9093");
    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(flinkConf);
    env.setParallelism(1);

    env.enableCheckpointing(2000);
    CheckpointConfig checkpointConfig = env.getCheckpointConfig();
    checkpointConfig.setMaxConcurrentCheckpoints(1);
    checkpointConfig.setCheckpointStorage("file:///flink_ckp");

    StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
    String ods1 = "CREATE TABLE ods_tb1 ( " +
        " stime STRING," +
        " name STRING," +
        " val BIGINT," +
        " rowtime AS to_timestamp(stime)," +
        " proctime as proctime()," +
        " WATERMARK FOR rowtime AS rowtime - interval '5' second" +
        ") WITH ( " +
        "  'connector' = 'kafka'," +
        "  'topic' = 'test1'," +
        "  'properties.bootstrap.servers' = 'kafka:9092'," +
        "  'properties.group.id' = 'JoinTest4'," +
        "  'scan.startup.mode' = 'latest-offset'," +
        "  'format' = 'json'" +
        ")";

    String ods2 = "CREATE TABLE ods_tb2 ( " +
        " stime STRING," +
        " name STRING," +
        " val BIGINT," +
        " rowtime AS to_timestamp(stime)," +
        " proctime as proctime()," +
        " WATERMARK FOR rowtime AS rowtime - interval '5' second" +
        ") WITH ( " +
        "  'connector' = 'kafka'," +
        "  'topic' = 'test2'," +
        "  'properties.bootstrap.servers' = 'kafka:9092'," +
        "  'properties.group.id' = 'JoinTest4'," +
        "  'properties.max.partition.fetch.bytes' = '10485760'," +
        "  'scan.startup.mode' = 'latest-offset'," +
        "  'format' = 'json'" +
        ")";


    String wind1 = "CREATE VIEW ww1 AS " +
        " SELECT " +
        " name,count(1) as cnt," +
        " window_start,window_end,window_time " +
        " from " +
        " table(tumble(table ods_tb1, descriptor(rowtime), interval '10' seconds))" +
        " GROUP BY name,window_start,window_end,window_time";

    String wind2 = "CREATE VIEW ww2 AS " +
        " SELECT " +
        " name,count(1) as cnt," +
        " window_start,window_end,window_time " +
        " from " +
        " table(tumble(table ods_tb2, descriptor(rowtime), interval '10' seconds))" +
        " GROUP BY name,window_start,window_end,window_time";


    String joinRes = "CREATE VIEW res AS " +
        " SELECT t1.name,t2.window_start as rowtime,t1.cnt as cnt1, t2.cnt as cnt2 FROM ww1 t1 " +
        " left join ww2 t2 " +
        " on t1.name = t2.name "
        + "AND t1.window_start=t2.window_start AND t1.window_end = t2.window_end"
        ;

    tableEnv.executeSql(ods1);
    tableEnv.executeSql(ods2);

    tableEnv.executeSql(wind1);
    tableEnv.executeSql(wind2);

    tableEnv.executeSql(joinRes);

    tableEnv.executeSql("desc res").print();

    tableEnv.executeSql("SELECT * FROM res").print();

  }
}
